package com.atguigu.day05;

import akka.pattern.BackoffSupervisor;
import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.IntSource;
import com.atguigu.utils.UrlCountPerWindow;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import scala.Int;

import java.util.ArrayList;
import java.util.List;

//全窗口聚合的实现 优化内存，只保留acc累加器，无需保留全部数据
//针对每个url 使用窗口进行分组
// ./home: {
//     window-1: acc,
//     window-2: acc
// }
public class Example2 {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .process(new FakeWindow(5000L))
                .print();


        env.execute();
    }

    //<key,输入，输出>
    public static class FakeWindow extends KeyedProcessFunction<String,ClickEvent,UrlCountPerWindow>{
        private long windowSize; //窗口长度

        public FakeWindow(long windowSize){
            this.windowSize = windowSize;
        }

        //key:WindowStartTime
        //value:窗口的累加器，Long类型
        private MapState<Long, Long> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, Long>(
                            "map-state",
                            Types.LONG,
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement(ClickEvent value, Context ctx, Collector<UrlCountPerWindow> out) throws Exception {
            //数据到达的机器时间
            long currTs = ctx.timerService().currentProcessingTime();
            // 如何计算currTs所属的窗口的开始时间？
            // 例如currTs是17，那么窗口开始时间是17-17%5 = 15
            long windowStartTime = currTs - currTs % windowSize;
            long windowEndTime = windowStartTime + windowSize;

            //将数据分配到窗口中
            if (!mapState.contains(windowStartTime)){
                //判断如果mapState状态变量中没有开始时间，说明窗口的第一条数据来了
                mapState.put(windowStartTime, 1L);
            } else {
                mapState.put(windowStartTime, mapState.get(windowStartTime) + 1L);
            }

            //当机器时间到达了窗口结束时间-1毫秒时，触发窗口的闭合计算，注册定时器
            ctx.timerService().registerProcessingTimeTimer(windowEndTime -1L);

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<UrlCountPerWindow> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            //聚合计算
            long windowEndTime = timestamp + 1L;
            long windowStartTime = windowEndTime - windowSize;
            String url = ctx.getCurrentKey();
            long count = mapState.get(windowStartTime); //在状态变量中找到当前key所对应的value就是累加起来的访问次数

            out.collect(new UrlCountPerWindow(url,count,windowStartTime,windowEndTime));
            //窗口触发计算以后，销毁窗口
            mapState.remove(windowStartTime);
        }
    }
    // {
//     "key-1": {window-1: acc, window-2: acc},
//     "key-2": {window-1: acc, window-2: acc}
// }


}

